/*
 * Proto Streamer
 * Copyright (C) 2009 John Pritchard.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License as
 * published by the Free Software Foundation; either version 2 of the
 * License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA.
 */
package p.s.m;

import com.google.protobuf.CodedInputStream;
import com.google.protobuf.CodedOutputStream;
import com.google.protobuf.Descriptors.FileDescriptor;
import com.google.protobuf.Descriptors.Descriptor;
import com.google.protobuf.Message;

import java.io.InputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.lang.reflect.Method;
import java.lang.reflect.InvocationTargetException;
import java.util.List;
import java.util.Map;

/**
 * Multiplex messages to I/O.  An application maintains one instance
 * of this class which is shared with all methods.
 * 
 * An application must register its message prototypes before doing
 * I/O.  It does this by calling the register method on its instance
 * of this class with the class generated by protoc.
 * 
 * @author jdp
 */
public final class Mux
    extends Object
{
    /**
     * Thrown from register method after consumer or producer have
     * entered.  At that point the mux connection has been initialized
     * and further registrations are not permitted.
     */
    public final static class Committed
        extends IllegalStateException
    {
        public Committed(){
            super();
        }
    }

    public final static class Binding 
        extends Object
    {
        public final static class Unknown
            extends IllegalStateException
        {
            public Unknown(int tag){
                super("Unknown binding tag '"+tag+"'.");
            }
            public Unknown(Message m){
                super("Unbound message type '"+m.getClass().getName()+"'");
            }
            public Unknown(Binding b, int tag){
                super("Unknown binding tag '"+tag+"' in '"+b.name+"'.");
            }
            public Unknown(Binding b, String type){
                super("Unknown binding for '"+type+"' in '"+b.name+"'.");
            }
        }
        public final static class BoundDescriptor
            extends Object
        {
            public final Descriptor messageDescriptor;
            public final Class messageClass;
            public final String name;
            public final int index;
            private final Method parseFrom;

            BoundDescriptor(Class proto, Descriptor type){
                super();
                this.messageDescriptor = type;
                try {
                    this.messageClass = Class.forName(proto.getName()+'$'+type.getName());
                }
                catch (ClassNotFoundException exc){
                    throw new IllegalStateException(exc);
                }
                this.name = this.messageClass.getName();
                this.index = type.getIndex();
                try {
                    this.parseFrom = this.messageClass.getMethod("parseFrom",CodedInputStream.class);
                }
                catch (NoSuchMethodException exc){
                    throw new IllegalStateException(this.messageClass.getName(),exc);
                }
            }

            public Message parseFrom(CodedInputStream in)
                throws IOException
            {
                try {
                    return (Message)this.parseFrom.invoke(null,in);
                }
                catch (InvocationTargetException iox){
                    throw (IOException)iox.getCause();
                }
                catch (IllegalAccessException exc){
                    throw new IllegalStateException(exc);
                }
            }
        }

        public final String name;
        public final Class protoClass;
        public final FileDescriptor protoDescriptor;
        public final int tag;
        public final List<BoundDescriptor> messageTypes;
        final byte[] lenDesc;


        Binding(Class protoClass, FileDescriptor protoDescriptor, int index){
            super();
            this.name = protoClass.getName();
            if (255 < this.name.length())
                throw new IllegalArgumentException("Proto class name '"+this.name+"' string length exceeded bindable limits");
            else {
                this.protoClass = protoClass;
                this.protoDescriptor = protoDescriptor;
                this.tag = index;
                {
                    String string = this.name;
                    int len = string.length();
                    byte[] buf = new byte[len+1];
                    this.name.getBytes(0,len,buf,1);
                    buf[0] = (byte)len;
                    this.lenDesc = buf;
                }
                List<Descriptor> messageTypes = protoDescriptor.getMessageTypes();
                if (255 < messageTypes.size())
                    throw new IllegalArgumentException("Proto '"+this.name+"' number of message types exceeded bindable limits");
                else {
                    this.messageTypes = new java.util.ArrayList<BoundDescriptor>();
                    for (Descriptor type: messageTypes){
                        this.messageTypes.add(new BoundDescriptor(protoClass,type));
                    }
                }
            }
        }

        public boolean in(Class clas){
            return (clas.equals(this.protoClass));
        }
        public String toString(){
            return this.name;
        }
        boolean validate(byte[] desc){
            byte[] lenDesc = this.lenDesc;
            if (lenDesc.length == desc.length+1){
                for (int cc = 0, count = desc.length; cc < count; cc++){
                    if (lenDesc[cc+1] != desc[cc])
                        return false;
                }
                return true;
            }
            else
                return false;
        }
        Message read(CodedInputStream in)
            throws IOException, Binding.Unknown
        {
            int tag = (in.readRawByte() & 0xff);
            BoundDescriptor md = this.messageTypes.get(tag);
            if (null != md)
                return md.parseFrom(in);
            else
                throw new Binding.Unknown(this,tag);
        }
        void write(Message m, Descriptor md, CodedOutputStream out)
            throws IOException
        {
            int tag = md.getIndex();
            out.writeRawByte(tag);
            m.writeTo(out);
            out.flush();
        }
    }

    private final static byte[] Head = {
        (byte)'m', (byte)'u', (byte)'x'
    };


    private final List<Binding> bindings;
    private final Map<Descriptor,Binding> bindingsMap;

    private volatile boolean committed;


    public Mux(){
        super();
        this.bindings = new java.util.ArrayList<Binding>();
        this.bindingsMap = new java.util.HashMap<Descriptor,Binding>();
    }


    /**
     * @param proto Class generated by protoc.
     */
    public void register(Class proto)
        throws IllegalArgumentException, IllegalStateException
    {
        if (this.committed)
            throw new Mux.Committed();
        try {
            Method m = proto.getMethod("getDescriptor");

            FileDescriptor protoDesc = (FileDescriptor)m.invoke(null);

            if (null != protoDesc)
                this.register(proto, protoDesc);
            else
                throw new IllegalArgumentException("Missing file descriptor in '"+proto.getName()+"'");
        }
        catch (Exception exc){
            throw new IllegalArgumentException(proto.getName(),exc);
        }
    }
    private void register(Class clas, FileDescriptor proto){
        List<Binding> bindings = this.bindings;

        int index = bindings.size();
        if (index < Byte.MAX_VALUE){
            for (Binding current : bindings){
                if (current.in(clas))
                    return;
            }
            Binding binding = new Binding(clas,proto,index);
            bindings.add(binding);

            Map<Descriptor,Binding> bindingsMap = this.bindingsMap;
            for (Binding.BoundDescriptor type: binding.messageTypes){
                bindingsMap.put(type.messageDescriptor,binding);
            }
        }
        else
            throw new IllegalArgumentException("Maximum number of bindings exceeded");
    }
    public void enter(Producer producer, CodedInputStream in)
        throws IOException
    {
        this.committed = true;
        /*
         * mux protocol head
         */
        byte[] head = in.readRawBytes(4);
        if ('m' == head[0] && 'u' == head[1] && 'x' == head[2]){
            List<Binding> bindings = this.bindings;
            /*
             * number of bindings
             */
            int count = (head[3] & 0xff);
            if (count == bindings.size()){
                /*
                 * bindings
                 */
                for (Binding binding: bindings){
                    int len = (in.readRawByte() & 0xff);
                    byte[] buf = in.readRawBytes(len);
                    if (!binding.validate(buf))
                        throw new IOException("Error in mux bindings order");
                }
            }
            else
                throw new IOException("Error in mux bindings number");
        }
        else
            throw new IOException("Missing mux protocol head");
    }
    public void enter(Consumer consumer, CodedOutputStream out)
        throws IOException
    {
        this.committed = true;
        int field = 0;
        /*
         * mux protocol head
         */
        out.writeRawBytes(Head);
        List<Binding> bindings = this.bindings;
        /*
         * number of bindings
         */
        out.writeRawByte(bindings.size());
        /*
         * bindings
         */
        for (Binding current : bindings){
            out.writeRawBytes(current.lenDesc);
        }
        out.flush();
    }
    public void exit(Consumer consumer){
        this.committed = false;
    }
    public void exit(Producer producer){
        this.committed = false;
    }
    public Message read(CodedInputStream in)
        throws IOException, Binding.Unknown
    {
        int tag = (in.readRawByte() & 0xff);
        Binding binding = this.bindings.get(tag);
        if (null != binding)
            return binding.read(in);
        else
            throw new Binding.Unknown(tag);
    }
    public void write(Message m, CodedOutputStream out)
        throws IOException, Binding.Unknown
    {
        Descriptor md = m.getDescriptorForType();
        Binding binding = this.bindingsMap.get(md);
        if (null != binding){
            out.writeRawByte(binding.tag);
            binding.write(m,md,out);
        }
        else
            throw new Binding.Unknown(m);
    }

}
